package com.zhangwoo.util;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobStatus;
import org.apache.hadoop.mapred.TaskReport;

public class HadoopJobMonitor {

	/**
	 * @param args
	 */
	public static void main(String[] args) {

	}

	public void checkStatus(){
		List<JobStatus> jobStatusList = new ArrayList<JobStatus>();

		try {
			Configuration conf = new Configuration();
			JobClient client = new JobClient(new JobConf(conf));
			JobStatus[] jobStatuses = client.getAllJobs();
			//	showFilter();

			int jobCount = 0;
			for (JobStatus jobStatus : jobStatuses) {
				Long lastTaskEndTime = 0L;
				TaskReport[] mapReports = client.getMapTaskReports(jobStatus.getJobID());
				for (TaskReport r : mapReports) {
					if (lastTaskEndTime < r.getFinishTime()) {
						lastTaskEndTime = r.getFinishTime();
					}
				}
				TaskReport[] reduceReports = client.getReduceTaskReports(jobStatus.getJobID());
				for (TaskReport r : reduceReports) {
					if (lastTaskEndTime < r.getFinishTime()) {
						lastTaskEndTime = r.getFinishTime();
					}
				}
				client.getSetupTaskReports(jobStatus.getJobID());
				client.getCleanupTaskReports(jobStatus.getJobID());

				String jobId = jobStatus.getJobID().toString();
				Long startTime = jobStatus.getStartTime();
				String user = jobStatus.getUsername();
				int mapProgress = (int)(jobStatus.mapProgress() * 100);
				int reduceProgress = (int)(jobStatus.reduceProgress() * 100);
				JobStatus jobStat = null;
				++jobCount;

				int runState = jobStatus.getRunState();
				switch(runState){
				case JobStatus.SUCCEEDED:
						Long duration = lastTaskEndTime - jobStatus.getStartTime();
					break;

				case JobStatus.RUNNING:
					break;

				case JobStatus.FAILED:
					break;

				case JobStatus.PREP:
					break;

				case JobStatus.KILLED:
					break;
				}

				jobStatusList.add(jobStat);
			}

			//get counters
			for (JobStatus jobStat : jobStatusList){
			}


//			showJobCounts();
		} catch (Exception ex) {
			System.out.println("Jobs status checker failed" + ex.getMessage());
		}

	}
}
